package parallel;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import event.Basic;

public class EventRangeInputFormat
	extends InputFormat<LongWritable, NullWritable> {

	/**
	 * An input split consisting of a range on numbers.
	 */

	/**
	 * A record reader that will generate a range of numbers.
	 */
	@Override
	public RecordReader<LongWritable, NullWritable> createRecordReader(
			InputSplit split,
			TaskAttemptContext context) throws IOException,
			InterruptedException {
		// TODO Auto-generated method stub
		return new BasicRangeRecordReader((BasicRangeInputSplit)split);
	}

	@Override
	public List<InputSplit> getSplits(
			JobContext context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		
        long totalsequences = getNumberOfsequences(context);
        int numSplits = getNumberOfRegions(context);
        long sequencesPerSplit = totalsequences / numSplits;
        System.out.println("Generating " + totalsequences + " using " + numSplits + 
                        " maps with step of " + sequencesPerSplit);
        List<InputSplit> splits = new ArrayList<InputSplit>();
        long currentsequence = 0;
        for(int split = 0; split < numSplits-1; ++split) {
                BasicRangeInputSplit inputSplit = new BasicRangeInputSplit(sequencesPerSplit, split);
                splits.add(inputSplit);
                currentsequence += sequencesPerSplit;
        }
        splits.add(new BasicRangeInputSplit(totalsequences - currentsequence, numSplits-1));
        for(int i = 0; i < splits.size(); i++)
        {
                System.out.println("splits id: "+ ((BasicRangeInputSplit)splits.get(i)).getSplitId());
        }
        return splits;
	}
	int getNumberOfRegions(JobContext context) {
		Configuration conf = context.getConfiguration();
		return conf.getInt("mapred.map.tasks", 1);
	}
	long getNumberOfsequences(JobContext context) {
		Configuration conf = context.getConfiguration();
		long sequences = conf.getLong("num-sequences", 0);
		return sequences;
	}

}
